package com.hr.reactive.hotAndcold;

import com.hr.reactive.common.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Flux;

import java.time.Duration;

public class Lec03AutoConnect {
    private static final Logger log = LoggerFactory.getLogger(Lec03AutoConnect.class);

    public static void main(String[] args) {
        //autoConnect 没有订阅者还在生产数据
       var movie = getMove().publish().autoConnect(2);
        Utils.sleepSeconds(2000);

        movie.take(1).subscribe(Utils.subscriber("sam"));
        Utils.sleepSeconds(3000);
        movie.take(2).subscribe(Utils.subscriber("mike"));
        Utils.sleepSeconds(10000);
    }

    private static Flux<String> getMove() {
        return Flux.generate(() -> {
                    log.info("received the request");
                    return 1;
                }, (state, sink) -> {
                    var sence = "movie sence " + state;
                    log.info("playing:{}", sence);
                    sink.next(sence);
                    return ++state;
                })
                .take(10)
                .delayElements(Duration.ofSeconds(1)).cast(String.class);

    }
}
